/* caPipeline.c */

/*
 * Licensed to the ConArtist Development Team (CADT).
 * See the NOTICE file distributed with this work for additional
 * information regarding copyright ownership.
 * The CADT licenses this file to you under the Apache License, 
 * Version 2.0 (the "License"); you may not use this file except in
 * compliance with the License.
 * You may obtain a copy of the License at
 * 
 *     http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <apr_hash.h>
#include <apr_thread_rwlock.h>
#include <apr_dso.h>
#include <apr_strings.h>

#include "conartist.h"
#include "caPipeline.h"
#include "caLog.h"
#include "caGenerator.h"

#include "caPrivate.h"

static apr_hash_t *pipeHash = NULL;
static apr_hash_t *pipelineHash = NULL;
static apr_thread_rwlock_t *modLock = NULL;

struct registeredPipe {
    const char *type;
    pipePrepareFn *func;
};

struct registeredPipeline {
    const char *name;
    struct caPipeline *pipeline;
};

apr_status_t pipeInit(void)
{
    pipeHash = apr_hash_make(globalPool);
    pipelineHash = apr_hash_make(globalPool);
    if (!pipeHash || ! pipelineHash)
        return APR_ENOMEM;
    return apr_thread_rwlock_create(&modLock, globalPool);
}

static apr_status_t findPipe(struct registeredPipe **rvp, 
                             const char *type)
{
    apr_status_t rv;
    *rvp = NULL;
    rv = apr_thread_rwlock_wrlock(modLock);
    if (rv == APR_SUCCESS) {
        *rvp = apr_hash_get(pipeHash, type, APR_HASH_KEY_STRING);
        apr_thread_rwlock_unlock(modLock);
        if (*rvp == NULL)
            rv = APR_EINVAL;
    }

    return rv;
}

/*
static apr_status_t findPipeline(struct registeredPipeline **rvp, 
                                 const char *name)
{
    apr_status_t rv;
    *rvp = NULL;
    rv = apr_thread_rwlock_wrlock(modLock);
    if (rv == APR_SUCCESS) {
        *rvp = apr_hash_get(pipelineHash, name, APR_HASH_KEY_STRING);
        apr_thread_rwlock_unlock(modLock);
        if (*rvp == NULL)
            rv = APR_EINVAL;
    }

    return rv;
}
*/

apr_status_t caPipeRegister(const char *type, pipePrepareFn *func)
{
    apr_status_t rv;
    struct registeredPipe *rp;

    rv = findPipe(&rp, type);
    if (rv == APR_SUCCESS) {
        caLog(CA_LOG_ERROR, "Attempt to register duplicate pipe name '%s'",
              type);
        return APR_EINVAL;
    }

    rp = apr_pcalloc(globalPool, sizeof(*rp));
    if (!rp)
        return APR_ENOMEM;
    rp->type = apr_pstrdup(globalPool, type);
    rp->func = func;

    rv = apr_thread_rwlock_wrlock(modLock);
    if (rv == APR_SUCCESS) {
        apr_hash_set(pipeHash, rp->type, APR_HASH_KEY_STRING, rp);
        apr_thread_rwlock_unlock(modLock);
        caLog(CA_LOG_INFO, "Registered pipe type '%s'", type);
    } else
        caLog(CA_LOG_WARN, "Unable to register pipe type '%s'", type);

    return rv;
}

apr_status_t caPipelineRegister(struct caPipeline *pl, const char *name)
{
    apr_status_t rv;
    struct registeredPipeline *rp = NULL;

    rp = apr_pcalloc(globalPool, sizeof(*rp));
    if (!rp)
        return APR_ENOMEM;
    rp->name = apr_pstrdup(globalPool, name);
    rp->pipeline = pl;

    rv = apr_thread_rwlock_wrlock(modLock);
    if (rv == APR_SUCCESS) {
        apr_hash_set(pipelineHash, rp->name, APR_HASH_KEY_STRING, rp);
        apr_thread_rwlock_unlock(modLock);
        caLog(CA_LOG_INFO, "Registered pipeline name '%s'", name);
    } else
        caLog(CA_LOG_WARN, "Unable to register pipeline name '%s'", name);

    return rv;
}

apr_status_t caPipeDataCreate(struct pipeData **piped, const char *type,
                              xmlNode *node)
{
    apr_status_t rv;
    struct registeredPipe *rp;
    struct pipeData *pd;

    *piped = NULL;
    if (!type)
        return APR_EINVAL;
    rv = findPipe(&rp, type);
    if (rv != APR_SUCCESS) {
        caLog(CA_LOG_ERROR, "Pipe type '%s' is not registered", type);
        return rv;
    }
    pd = apr_pcalloc(globalPool, sizeof(*pd));
    if (! pd)
        return APR_ENOMEM;

    pd->cfgNodes = xmlCopyNode(node, 1); /* recursive copy */
    pd->func = *rp->func;

    *piped = pd;
    return APR_SUCCESS;
}

apr_status_t caPipelineCreate(struct caPipeline **pl, const char *name,
                              apr_pool_t *p)
{
    struct caPipeline *pipeline = NULL;

    if (!p)
        return APR_ENOPOOL;

    pipeline = apr_pcalloc(p, sizeof(*pipeline));
    if (!pipeline)
        return APR_ENOMEM;

    pipeline->name = apr_pstrdup(p, name);
    pipeline->pool = p;
    *pl = pipeline;
    return APR_SUCCESS;
}

apr_status_t caPipelineInstanceCreatePipes(struct pipelineInstance *pi)
{
    struct caPipe **array = NULL;
    int i;
    apr_status_t rv = APR_SUCCESS;

    if (!pi)
        return APR_EINVAL;

    array = apr_pcalloc(pi->gen->pool, 
                        sizeof(*array) * pi->pipeline->nPipes);
    if (!array)
        return APR_EINVAL;

    for (i = 0; i < pi->pipeline->nPipes; i++) {
        struct pipeData *pd = pi->pipeline->pipeData[i];
        rv = (*pd->func)(&array[i], pi, pd->cfgNodes);
        if (rv != APR_SUCCESS)
            break;
    }
    if (rv == APR_SUCCESS)
        pi->pipes = array;

    return rv;
}

apr_status_t caPipelineAddPipeData(struct caPipeline *pl, 
                                   struct pipeData *pd)
{
    struct pipeData **array = NULL;
    int nSz = pl->nPipes + 1;

    if (!pl || !pd)
        return APR_EINVAL;

    array = apr_pcalloc(pl->pool, sizeof(*array) * nSz);
    if (!array)
        return APR_EINVAL;

    if (pl->nPipes)
        memcpy(array, pl->pipeData, sizeof(*array) * pl->nPipes);

    array[pl->nPipes++] = pd;
    pl->pipeData = array;

    return APR_SUCCESS;
}

apr_status_t caPipelineRun(struct caPipeData *pd, 
                           struct pipelineInstance *pi)
{
    int i;
    apr_status_t rv;

    pd->output = NULL;
    pd->oLen = 0;

    if (!pi->pipes) {
        caLog(CA_LOG_WARN, "Pipeline instance with no pipes!");
        return APR_EINVAL;
    }

    for (i = 0; i < pi->pipeline->nPipes; i++) {
        struct caPipe *pipe = pi->pipes[i];

        /* swap output to input */
        if ((pipe->flags & CAP_NOSWAP) == 0 && i && pd->oLen > 0) {
            pd->input = pd->output;
            pd->iLen = pd->oLen;
            pd->output = NULL;
            pd->oLen = 0;
        }

        rv = (pipe->func)(pd, pi->pipes[i]);
        if (rv != pipe->rValue)
            return APR_EINVAL; /* TODO - return value? */
    }

    return APR_SUCCESS;
}
